package com.wuwangfu.window.reduce;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: jcshen
 * @Date: 2023-03-08
 * @PackageName: com.wuwangfu.window.reduce
 * @ClassName: WindowReduceFunction
 * @Description: 实现窗口聚合+状态聚合
 * @Version: 1.0.0
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/windows/#reducefunction
 *
 */
public class WindowReduceFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        1000,hive,1
//        2000,hive,1
//        5000,hive,1

//        6000,flink,2
//        7000,flink,2
//        9000,hive,2
//        10000,flink,1
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //生成watermark
        SingleOutputStreamOperator<String> dataWatermark = lines.assignTimestampsAndWatermarks(WatermarkStrategy
                .<String>forBoundedOutOfOrderness(Duration.ZERO)
                //提取时间戳
                .withTimestampAssigner((line, timstamp) -> Long.parseLong(line.split(",")[0]))

        );
        //
        SingleOutputStreamOperator<Tuple2<String, Integer>> maped = dataWatermark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });
        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyed = maped.keyBy(t -> t.f0);
        //开窗
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        //如果直接调用sum 或 reduce，只会聚合窗口内的数据，不去跟历史数据进行累加
        /*需求：可以在窗口内进行增量聚合，并且还可以与历史数据进行聚合*/
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowed.reduce(new MyReduceFunc(), new MyWindowFunc());

        result.print();


        env.execute();
    }

    private static class MyReduceFunc implements ReduceFunction<Tuple2<String, Integer>> {
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> v1, Tuple2<String, Integer> t1) throws Exception {
            v1.f1 = v1.f1 + t1.f1;
            return v1;
        }
    }

    private static class MyWindowFunc extends ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> {

        private transient ValueState<Integer> valueState;

        @Override
        public void open(Configuration parameters) throws Exception {

            ValueStateDescriptor<Integer> valueDesc = new ValueStateDescriptor<>("wc-state", Integer.class);
            valueState = getRuntimeContext().getState(valueDesc);
        }

        @Override
        public void process(String key, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<Tuple2<String, Integer>> out) throws Exception {
           //获取当前窗口的起止时间和水位线
            //long start = context.window().getStart();
            //long watermark = context.currentWatermark();

            Integer his = valueState.value();
            if (his == null) {
                his = 0;
            }
            //当前窗口数据
            Tuple2<String, Integer> tp = iterable.iterator().next();
            //更新状态
            valueState.update(tp.f1 + his);
            //窗口数据 + 状态数据
            tp.f1 = tp.f1 + his;
            //输出
            out.collect(tp);

        }
    }


//    private static class MyWindowFunc implements WindowFunction<Tuple2<String,Integer>,Tuple2<String,Integer>,String,TimeWindow> {
//        @Override
//        public void apply(String key, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) {
//
//        }
//    }
}

